#include "ThreadPool.h"
#include <iostream>
#include <sstream>
#include <algorithm>

namespace common_component
{

    ThreadPool::ThreadPool(size_t num_threads, const std::string &name)
        : name_(name), stop_(false), active_threads_(0)
    {
        if (num_threads == 0)
        {
            num_threads = std::thread::hardware_concurrency();
            if (num_threads == 0)
            {
                num_threads = 4;
            }
        }
        std::cout << "[ThreadPool] Creating thread pool " << name << " with " << num_threads << " threads" << std::endl;
        stats_ = Statistics{};
        stats_.total_tasks_executed = 0;
        stats_.total_tasks_failed = 0;
        stats_.average_wait_time_ms = 0.0;
        stats_.average_execution_time_ms = 0.0;
        stats_.max_queue_size = 0;
        // 创建工作线程
        for (size_t i = 0; i < num_threads; ++i)
        {
            workers_.emplace_back([this]()
                                  { WorkerThread(); });
        }
        std::cout << "[ThreadPool] Thread pool " << name_ << " created successfully" << std::endl;
    }
    ThreadPool::~ThreadPool()
    {
        {
            std::unique_lock<std::mutex> lock(queue_mutex_);
            if (stop_)
            {
                return;
            }
            stop_ = true;
        }
        std::cout << "[ThreadPool] Shutting down thread pool " << name_ << "...." << std::endl;
        condition_.notify_all();
        for (std::thread &worker : workers_)
        {
            if (worker.joinable())
            {
                worker.join();
            }
        }
        workers_.clear();
        std::priority_queue<Task> empty;
        std::swap(tasks_, empty);
        std::cout << "[ThreadPool] Thread pool " << name_ << " shutdown complete!" << std::endl;
    }
    void ThreadPool::Resize(size_t num_threads)
    {
        std::cout << "[ThreadPool] Dynamic resize not yet implemented" << std::endl;
    }
    void ThreadPool::WaitAll()
    {
        std::unique_lock<std::mutex> lock(queue_mutex_);
        finish_condition_.wait(lock, [this]()
                               { return tasks_.empty() && active_threads_ == 0; });
    }

    void ThreadPool::Shutdown(bool wait_for_completion)
    {
        std::cout << "[ThreadPool] Shutting down thread pool " << name_ << "..." << std::endl;

        {
            std::unique_lock<std::mutex> lock(queue_mutex_);

            // 如果已经关闭，直接返回
            if (stop_)
            {
                std::cout << "[ThreadPool] Thread pool " << name_ << " already shut down" << std::endl;
                return;
            }

            // 如果需要等待任务完成
            if (wait_for_completion)
            {
                std::cout << "[ThreadPool] Waiting for " << tasks_.size()
                          << " pending tasks to complete..." << std::endl;
                // 等待所有任务完成
                finish_condition_.wait(lock, [this]
                                       { return tasks_.empty() && active_threads_ == 0; });
            }
            else
            {
                // 清空任务队列
                std::priority_queue<Task> empty;
                std::swap(tasks_, empty);
                std::cout << "[ThreadPool] Cleared pending tasks" << std::endl;
            }

            // 设置停止标志
            stop_ = true;
        }

        // 通知所有工作线程退出
        condition_.notify_all();

        // 等待所有线程结束
        for (std::thread &worker : workers_)
        {
            if (worker.joinable())
            {
                worker.join();
            }
        }

        // 清理资源
        workers_.clear();

        std::cout << "[ThreadPool] Thread pool " << name_ << " shutdown complete!" << std::endl;
    }

    void ThreadPool::Clear()
    {
        std::unique_lock<std::mutex> lock(queue_mutex_);
        std::priority_queue<Task> empty;
        std::swap(tasks_, empty);
    }
    void ThreadPool::SetThreadAffinity(size_t thread_index, size_t cpu_index)
    {
        // 简化实现：暂不支持CPU亲和性设置
        std::cout << "[ThreadPool] Thread affinity not yet implemented" << std::endl;
    }
    void ThreadPool::WorkerThread()
    {
        static std::atomic<int> thread_counter{0};
        int thread_id = thread_counter++;

        std::cout << "[ThreadPool] Worker thread " << thread_id << " started in pool '" << name_ << "'"
                  << std::endl;

        // 设置线程名称（用于调试）
#ifdef _WIN32
        // Windows线程命名
        std::ostringstream oss;
        oss << name_ << "_Worker" << thread_id;
        std::string thread_name = oss.str();
        // SetThreadDescription() 需要Windows 10 1607+
#else
        // Linux线程命名
        std::ostringstream oss;
        oss << name_ << "_W" << thread_id;
        std::string thread_name = oss.str();
        if (thread_name.length() > 15)
        {
            thread_name = thread_name.substr(0, 15); // pthread名称限制为15字符
        }
        pthread_setname_np(pthread_self(), thread_name.c_str());
#endif

        while (true)
        {
            Task task(Priority::NORMAL, nullptr);
            bool has_task = false;

            {
                std::unique_lock<std::mutex> lock(queue_mutex_);

                // 等待任务或停止信号
                condition_.wait(lock, [this]
                                { return stop_ || !tasks_.empty(); });

                if (stop_ && tasks_.empty())
                {
                    break;
                }

                if (!tasks_.empty())
                {
                    // 从优先队列获取任务
                    task = std::move(const_cast<Task &>(tasks_.top()));
                    tasks_.pop();
                    has_task = true;
                    active_threads_++;

                    // 更新统计
                    auto now = std::chrono::steady_clock::now();
                    auto wait_time =
                        std::chrono::duration_cast<std::chrono::milliseconds>(now - task.submit_time)
                            .count();

                    // 增量平均值计算公式：
                    // 新平均值 = (旧平均值 × 旧样本数 + 新样本值) / 新样本数
                    //
                    // 数学原理：
                    // 假设已有n个样本，平均值为avg_n，现在加入第(n+1)个样本x_{n+1}
                    // 旧的总和 = avg_n × n
                    // 新的总和 = avg_n × n + x_{n+1}
                    // 新平均值 = 新总和 / (n+1) = (avg_n × n + x_{n+1}) / (n+1)
                    //
                    // 优势：无需存储所有历史数据，只需要当前平均值和样本数
                    stats_.average_wait_time_ms =
                        (stats_.average_wait_time_ms * stats_.total_tasks_executed + wait_time) /
                        (stats_.total_tasks_executed + 1);
                }
            }

            // 执行任务（不持有锁）
            if (has_task && task.func)
            {
                auto start_time = std::chrono::steady_clock::now();

                try
                {
                    task.func();
                    stats_.total_tasks_executed++;
                }
                catch (const std::exception &e)
                {
                    std::cerr << "[ThreadPool] Worker " << thread_id << " task exception: " << e.what()
                              << std::endl;
                    stats_.total_tasks_failed++;
                }
                catch (...)
                {
                    std::cerr << "[ThreadPool] Worker " << thread_id << " unknown exception"
                              << std::endl;
                    stats_.total_tasks_failed++;
                }

                auto end_time = std::chrono::steady_clock::now();
                auto exec_time =
                    std::chrono::duration_cast<std::chrono::milliseconds>(end_time - start_time)
                        .count();
                stats_.average_execution_time_ms =
                    (stats_.average_execution_time_ms * (stats_.total_tasks_executed - 1) + exec_time) /
                    stats_.total_tasks_executed;
                active_threads_--;
                // 任务完成后，需要在锁保护下更新active_threads_
                {
                    std::lock_guard<std::mutex> lock(queue_mutex_);

                    // 检查是否所有任务都完成了
                    if (tasks_.empty() && active_threads_ == 0)
                    {
                        finish_condition_.notify_all();
                    }
                }
            }
        }

        std::cout << "[ThreadPool] Worker thread " << thread_id << " stopped in pool '" << name_ << "'"
                  << std::endl;
    }

    // 为了兼容StreamingEngine中使用的简化接口，添加enqueue函数
    auto ThreadPool::enqueue(std::function<void()> f) -> std::future<void>
    {
        return Submit(Priority::NORMAL, f);
    }

} // namespace common_component